package main

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/gin-gonic/gin"
)

type Controller struct {
	stores          StoreSet
	tablesDefs      map[string]map[string]TableDefine
	mu              sync.RWMutex
	interval        int
	stopping        chan interface{}
	stopped         chan interface{}
	o, r, w         int64
	tableDefineFile string
	storeBasePath   string
}

func NewController(defpath, storepath string, interval int) (*Controller, error) {
	c := &Controller{
		stopping:        make(chan interface{}),
		stopped:         make(chan interface{}),
		tableDefineFile: defpath,
		storeBasePath:   storepath,
		interval:        *IntervalToCloseStore,
		o:               *OpenDuration,
		r:               *ReadDuration,
		w:               *WriteDuation,
	}
	if err := c.LoadTableDef(); err != nil {
		return nil, err
	}
	if err := c.LoadLocalData(); err != nil {
		return nil, err
	}
	go c.CheckAndClose()
	return c, nil
}

func (c *Controller) Close() {
	close(c.stopping)
	<-c.stopped
}

func (c *Controller) CloseStore(closeAll bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	ref := c.stores
	if !closeAll {
		ref = c.stores.FilterByActive(c.o, c.r, c.w)
	}
	for _, s := range ref {
		if s.IsActive() {
			s.Close()
		}
	}
}

func (c *Controller) CheckAndClose() {
	tick := time.Tick(time.Second * time.Duration(c.interval))
	for {
		select {
		case <-tick:
			c.CloseStore(false)
		case <-c.stopping:
			c.CloseStore(true)
			close(c.stopped)
		}
	}
}

func (c *Controller) LoadTableDef() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	_, err := os.Stat(c.tableDefineFile)
	if os.IsNotExist(err) {
		file, err := os.OpenFile(c.tableDefineFile, os.O_CREATE|os.O_RDWR, 0755)
		if err != nil {
			return err
		}
		if _, err := file.Write([]byte("[]")); err != nil {
			return err
		}
		file.Close()
	}
	b, err := ioutil.ReadFile(c.tableDefineFile)
	if err != nil {
		return err
	}
	var defs []TableDefine
	if err := json.Unmarshal(b, &defs); err != nil {
		return err
	}
	c.tablesDefs = make(map[string]map[string]TableDefine)
	for _, df := range defs {
		if _, ok := c.tablesDefs[df.Appid]; !ok {
			c.tablesDefs[df.Appid] = make(map[string]TableDefine)
		}
		if _, ok := c.tablesDefs[df.Appid][df.TableName]; ok {
			return fmt.Errorf("duplicated table name %s", df.TableName)
		}
		c.tablesDefs[df.Appid][df.TableName] = df
	}
	return nil
}

func (c *Controller) WalkLoad(path string, info os.FileInfo, err error) error {
	if !info.IsDir() {
		return nil
	}
	p, err := filepath.Rel(c.storeBasePath, path)
	if err != nil {
		return err
	}
	p = strings.TrimSuffix(p, "/")
	p = strings.TrimPrefix(p, "/")
	ss := strings.Split(p, "/")
	if len(ss) != 3 {
		return nil
	}
	appid := ss[0]
	tablename := ss[1]
	partition := ss[2]
	if appid == "" || tablename == "" || partition == "" {
		return nil
	}
	par, err := NewPartition(partition)
	if err != nil {
		return nil
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, ok := c.tablesDefs[appid]; !ok {
		return nil
	}
	if _, ok := c.tablesDefs[appid][tablename]; !ok {
		return nil
	}
	def := c.tablesDefs[appid][tablename]
	if def.DtLength != len(par.dt) {
		return fmt.Errorf("dt length not match %s", path)
	}
	store, err := DefaultStoreFunc(def, *par, c.storeBasePath)
	if err != nil {
		return err
	}
	c.stores = append(c.stores, store)
	return nil
}

func (c *Controller) LoadLocalData() error {
	return filepath.Walk(c.storeBasePath, c.WalkLoad)
}

func (c *Controller) createTableHandler(ctx *gin.Context) {
	t := time.Now().UnixNano() / 1000 / 1000
	var err error
	appid := ctx.Param("appid")
	tablename := ctx.Param("tablename")
	dtLength, err := strconv.Atoi(ctx.PostForm("dt_length"))
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(400, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	index := ctx.PostForm("index")
	var idxcols []string
	if err = json.Unmarshal([]byte(index), &idxcols); err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, ok := c.tablesDefs[appid][tablename]; ok {
		// ctx.AbortWithError(403, fmt.Errorf("table define already exists"))
		ctx.JSON(403, gin.H{
			"desc": "table define already exists",
			"status": "error",
		})
		return
	}
	if _, ok := c.tablesDefs[appid]; !ok {
		c.tablesDefs[appid] = make(map[string]TableDefine)
	}
	c.tablesDefs[appid][tablename] = TableDefine{
		Appid:        appid,
		TableName:    tablename,
		DtLength:     dtLength,
		IndexColumns: idxcols,
	}
	b, err := ioutil.ReadFile(c.tableDefineFile)
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	var defs []TableDefine
	if err = json.Unmarshal(b, &defs); err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	defs = append(defs, c.tablesDefs[appid][tablename])
	if b, err = json.Marshal(defs); err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	if err = ioutil.WriteFile(c.tableDefineFile, b, 0755); err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	delta := time.Now().UnixNano()/1000/1000 - t
	ctx.JSON(200, gin.H{
		"cost_ms": delta,
		"status":  "ok",
	})
}

func (c *Controller) getTableHandler(ctx *gin.Context) {
	def, err := c.getDefine(ctx)
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(404, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	b, err := json.Marshal(def)
	if err != nil {
		// ctx.AbortWithError(500, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	ctx.JSON(200, b)
}

func (c *Controller) getDefine(ctx *gin.Context) (*TableDefine, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	appid := ctx.Param("appid")
	tablename := ctx.Param("tablename")
	m, ok := c.tablesDefs[appid]
	if !ok {
		return nil, fmt.Errorf("appid %s not found", appid)
	}
	def, ok := m[tablename]
	if !ok {
		return nil, fmt.Errorf("tablename %s not found", tablename)
	}
	return &def, nil
}

func (c *Controller) writeHandler(ctx *gin.Context) {
	t := time.Now().UnixNano() / 1000 / 1000
	def, err := c.getDefine(ctx)
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(404, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	dt := ctx.PostForm("dt")
	partition := ctx.PostForm("partition")
	data := ctx.PostForm("data")
	if len(dt) != def.DtLength {
		// ctx.AbortWithError(404, fmt.Errorf("dt length not match define"))
		ctx.JSON(404, gin.H{
			"desc": "dt length not match define",
			"status": "error",
		})
		return
	}
	var rows []Row
	if err := json.Unmarshal([]byte(data), &rows); err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(400, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	store, ok, err := c.stores.GetOrCreate(*def, Partition{partition: partition, dt: dt}, c.storeBasePath)
	if ok {
		c.stores = c.stores.MustAdd(store)
	}
	if err := json.Unmarshal([]byte(data), &rows); err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(400, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	if !store.IsActive() {
		if err := store.Open(); err != nil {
			// ctx.AbortWithError(500, err)
			ctx.JSON(500, gin.H{
				"desc": err.Error(),
				"status": "error",
			})
			return
		}
	}
	rowpts := make([]*Row, len(rows))
	for i, rr := range rows {
		rowpts[i] = &rr
	}
	if err := store.Write(rowpts); err != nil {
		// ctx.AbortWithError(500, err)
		ctx.JSON(500, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	delta := time.Now().UnixNano()/1000/1000 - t
	ctx.JSON(200, gin.H{
		"cost_ms": delta,
		"status":  "ok",
	})
}

func (c *Controller) readGetHandler(ctx *gin.Context) {
	t := time.Now().UnixNano() / 1000 / 1000
	def, err := c.getDefine(ctx)
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(404, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	key := ctx.Param("Key")
	dtPrefix := ctx.Query("dt")
	partition := ctx.Query("partition")
	c.mu.RLock()
	defer c.mu.RUnlock()
	ss := c.stores.FilterByTable(def.Appid, def.TableName).FilterByPartition(partition, dtPrefix)
	var ret []RowData
	for _, s := range ss {
		if !s.IsActive() {
			if err := s.Open(); err != nil {
				// ctx.AbortWithError(500, err)
				ctx.JSON(500, gin.H{
					"desc": err.Error(),
					"status": "error",
				})
				return
			}
		}
		value, err := s.Query(key)
		if err != nil {
			// ctx.AbortWithError(500, err)
			ctx.JSON(404, gin.H{
				"desc": err.Error(),
				"status": "error",
			})
			return
		}
		rd := RowData{
			Partition: s.GetPartition().partition,
			Dt:        s.GetPartition().dt,
			Result:    map[string]string{key: value},
		}
		ret = append(ret, rd)
	}
	delta := time.Now().UnixNano()/1000/1000 - t
	ctx.JSON(200, gin.H{
		"cost_ms": delta,
		"data":    ret,
	})
}

func (c *Controller) readPostHandler(ctx *gin.Context) {
	t := time.Now().UnixNano() / 1000 / 1000
	def, err := c.getDefine(ctx)
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(404, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	key_prefix := ctx.PostForm("key_prefix")
	column_prefix := ctx.PostForm("column_prefix")
	column_equal := ctx.PostForm("column_equal")
	key_only := ctx.PostForm("key_only") != ""
	dtPrefix := ctx.PostForm("dt")
	partition := ctx.PostForm("partition")
	cond := QueryCondition{
		keyPrefix: key_prefix,
	}
	var idx map[string]string
	switch {
	case column_prefix != "" && column_equal == "":
		err := json.Unmarshal([]byte(column_prefix), &idx)
		if err != nil {
			// ctx.AbortWithError(404, err)
			ctx.JSON(404, gin.H{
				"desc": err.Error(),
				"status": "error",
			})
			return
		}
		cond.index = idx
		cond.indexPrefixQuery = true
	case column_equal != "" && column_prefix == "":
		err := json.Unmarshal([]byte(column_equal), &idx)
		if err != nil {
			// ctx.AbortWithError(404, err)
			ctx.JSON(404, gin.H{
				"desc": err.Error(),
				"status": "error",
			})
			return
		}
		cond.index = idx
		cond.indexPrefixQuery = false
	default:
		//ctx.AbortWithError(403, fmt.Errorf("column_prefix column_equal conflict"))
		ctx.JSON(400, gin.H{
			"desc": "column_prefix column_equal confilct",
			"status": "error",
		})
		return
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	ss := c.stores.FilterByTable(def.Appid, def.TableName).FilterByPartition(partition, dtPrefix)
	var ret []RowData
	for _, s := range ss {
		if !s.IsActive() {
			if err := s.Open(); err != nil {
				// ctx.AbortWithError(500, err)
				ctx.JSON(500, gin.H{
					"desc": err.Error(),
					"status": "error",
				})
				return
			}
		}
		if key_only {
			keys, err := s.QueryKeys(cond)
			if err != nil {
				// ctx.AbortWithError(500, err)
				ctx.JSON(404, gin.H{
					"desc": err.Error(),
					"status": "error",
				})
				return
			}
			rd := RowData{
				Partition: s.GetPartition().partition,
				Dt:        s.GetPartition().dt,
				Keys:      keys,
			}
			if keys != nil {
				ret = append(ret, rd)
			}
		} else {
			values, err := s.QueryValues(cond)
			if err != nil {
				// ctx.AbortWithError(500, err)
				ctx.JSON(404, gin.H{
					"desc": err.Error(),
					"status": "error",
				})
				return
			}
			rd := RowData{
				Partition: s.GetPartition().partition,
				Dt:        s.GetPartition().dt,
				Result:    values,
			}
			if values != nil {
				ret = append(ret, rd)
			}
		}
	}
	delta := time.Now().UnixNano()/1000/1000 - t
	ctx.JSON(200, gin.H{
		"cost_ms": delta,
		"data":    ret,
	})
}

func (c *Controller) deleteHandler(ctx *gin.Context) {
	t := time.Now().UnixNano() / 1000 / 1000
	def, err := c.getDefine(ctx)
	if err != nil {
		// ctx.AbortWithError(404, err)
		ctx.JSON(404, gin.H{
			"desc": err.Error(),
			"status": "error",
		})
		return
	}
	key := ctx.Param("key")
	dt := ctx.Query("dt")
	partition := ctx.Query("partition")
	ss := c.stores.FilterByTable(def.Appid, def.TableName).FilterByPartition(partition, dt)
	for _, s := range ss {
		if !s.IsActive() {
			if err := s.Open(); err != nil {
				// ctx.AbortWithError(500, err)
				ctx.JSON(500, gin.H{
					"desc": err.Error(),
					"status": "error",
				})
				return
			}
		}
		if err = s.Delete(key); err != nil {
			// ctx.AbortWithError(404, err)
			ctx.JSON(404, gin.H{
				"desc": err.Error(),
				"status": "error",
			})
			return
		}
	}
	delta := time.Now().UnixNano()/1000/1000 - t
	ctx.JSON(200, gin.H{
		"cost_ms": delta,
		"status":  "ok",
	})
}
